package com.diego.strom.demo.wordcount;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.util.Map;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

/**
 * @author Manfred
 *	1.创建一个Spout作为数据源
    Spout作为数据源，它实现了IRichSpout接口，功能是读取一个文本文件并把它的每一行内容
 */

public class WordRichSpout implements IRichSpout {

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	private SpoutOutputCollector collector;

	private FileReader fileReader;

	private boolean completed = false;

	public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
		try {
			// 获取创建Topology时指定的要读取的文件路径
			this.fileReader = new FileReader(conf.get("wordsFile").toString());
		} catch (FileNotFoundException e) {
			throw new RuntimeException("Error reading file [" + conf.get("wordFile") + "]");
		}
		// 初始化发射器
		this.collector = collector;

	}

	public void close() {
		// TODO Auto-generated method stub

	}

	public void activate() {
		// TODO Auto-generated method stub

	}

	public void deactivate() {
		// TODO Auto-generated method stub

	}

	public void nextTuple() {
		if (completed) {
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				// Do nothing
			}
			return;
		}
		String str;
		// Open the reader
		BufferedReader reader = new BufferedReader(fileReader);
		try {
			// Read all lines
			while ((str = reader.readLine()) != null) {
				/**
				 * 发射每一行，Values是一个ArrayList的实现
				 */
				this.collector.emit(new Values(str), str);
			}
		} catch (Exception e) {
			throw new RuntimeException("Error reading tuple", e);
		} finally {
			completed = true;
		}

	}

	public void ack(Object msgId) {
		System.out.println("OK:" + msgId);  
	}

	public void fail(Object msgId) {
		System.out.println("FAIL:" + msgId); 
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		  declarer.declare(new Fields("line"));  

	}

	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		return null;
	}

}
